package transport;

import java.io.IOException;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;

/**
 * UniProtocolAgent class provides pipe protocol.
 */
public final class UniProtocolAgent extends ProtocolAgent {
    private final UniProtocol protocol;
    public UniProtocolAgent(UniProtocol protocol, Socket socket) throws IOException {
        super(protocol, socket);

        this.protocol = protocol;
    }

    public UniProtocol getProtocol() {
        return protocol;
    }

    public static String readString(ByteBuffer buffer) {
        short length = buffer.getShort();
        CharBuffer charBuf = buffer.asCharBuffer();
        return String.valueOf(charBuf.array(), charBuf.arrayOffset() + charBuf.position(), length);
    }

    public static void writeString(ByteBuffer buffer, String value) {
        buffer.putShort((short)value.length());
        buffer.asCharBuffer().put(value);
    }

    @Override
    protected synchronized void processMessage(ByteBuffer message) {
        String queueName = readString(message);

        UniClient [] clients = protocol.getQueueClients(queueName);

        message.mark();
        for (UniClient client : clients) {
            message.reset();
            client.processMessage(this, queueName, message);
        }
    }

    public final class Message {
        public final String queueName;
        public final List<ByteBuffer> buffers = new LinkedList<ByteBuffer>();

        private ByteBuffer current;

        public Message(String queueName) {
            this.queueName = queueName;
            current = protocol.buffersCache.allocate();
            buffers.add(current);
        }

        private void ensure(int size) {
            assert size <= BUFFER_SIZE;

            if (current != null && current.remaining() >= size) {
                return ;
            }

            current = protocol.buffersCache.allocate();
            buffers.add(current);
        }

        public void putShort(short value) {
            ensure(2);
            current.putShort(value);
        }

        public void putInt(int value) {
            ensure(4);
            current.putInt(value);
        }

        public void putString(String value) {
            ensure(2 + value.length() * 2);

            writeString(current, value);
        }

        public void put(ByteBuffer data) {
            // message was flipped
            assert current != null;

            int offset = data.arrayOffset() + data.position();
            int remain = data.remaining();

            while (remain > 0) {
                int toCopy = Math.min(remain, current.remaining());
                current.put(data.array(), offset, toCopy);

                offset += toCopy;
                remain -= toCopy;

                if (remain > 0) {
                    current.flip();

                    current = protocol.buffersCache.allocate();
                    buffers.add(current);
                }
            }
        }

        public void flip() {
            if (current != null) {
                current.flip();
                current = null;
            }
        }

        public void release() {
            for (ByteBuffer buffer : buffers) {
                protocol.buffersCache.release(buffer);
            }

            current = null;
            buffers.clear();
        }
    }

    private final Queue<Message> messageQueue = new LinkedList<Message>();

    @Override
    protected void processWrite() throws IOException {
        synchronized (messageQueue) {
            if (messageQueue.isEmpty()) {
                return ;
            }
        }

        for (;;) {
            Message msg;

            synchronized (messageQueue) {
                if (messageQueue.isEmpty()) {
                    return ;
                }

                msg = messageQueue.remove();
            }

            msg.flip();
            sendMessage(msg.buffers);
            msg.release();
        }
    }

    public final void sendMessage(String queueName, ByteBuffer data) {
        Message msg = new Message(queueName);
        msg.putString(queueName);
        msg.put(data);
        msg.flip();

        synchronized (messageQueue) {
            messageQueue.add(msg);
        }
    }

    public final Message beginMessage(String queueName) {
        return new Message(queueName);
    }

    public final void endMessage(Message msg) {
        msg.flip();

        synchronized (messageQueue) {
            messageQueue.add(msg);
        }
    }
}
